package com.tt.job.service;

import com.tt.job.api.JobContext;
import com.tt.job.api.schedule.JobSchedule;
import com.tt.job.api.schedule.ScheduleType;
import com.tt.job.api.schedule.async.AsyncJobSchedule;
import com.tt.job.api.schedule.event.EventJobSchedule;
import com.tt.job.api.schedule.triger.TriggerJobSchedule;
import com.tt.job.api.service.JobProduceApi;
import com.tt.job.quartz.TriggerJobEventProduceJob;
import com.tt.job.register.JobRegistryCenter;
import com.tt.job.register.ScheduleChangeListener;
import com.tt.job.rocketmq.MqProducer;
import org.apache.commons.lang3.StringUtils;
import org.quartz.CronScheduleBuilder;
import org.quartz.JobBuilder;
import org.quartz.JobDataMap;
import org.quartz.JobDetail;
import org.quartz.JobKey;
import org.quartz.ScheduleBuilder;
import org.quartz.Scheduler;
import org.quartz.SchedulerException;
import org.quartz.SchedulerFactory;
import org.quartz.Trigger;
import org.quartz.TriggerBuilder;
import org.quartz.impl.StdSchedulerFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * 事件生产者服务，实现作为leader时，发起任务执行消息
 * 顺带着进行schedule信息的维护工作
 */
@Service
public class JobEventProduceService implements JobProduceApi, ScheduleChangeListener {
    private final static Logger logger = LoggerFactory.getLogger(JobEventProduceService.class);
    private final SchedulerFactory schedulerFactory = new StdSchedulerFactory();
    @Autowired
    MqProducer mqProducer;
    @Autowired
    JobRegistryCenter registryCenter;
    /**
     * 异步调度任务 group#jobClass -> schedule
     */
    private ConcurrentHashMap<String, AsyncJobSchedule> asyncSchedules = new ConcurrentHashMap<>();
    /**
     * 事件调度信息维护  group#event -> List&lt;Schedule&gt;
     */
    private ConcurrentHashMap<String, Set<EventJobSchedule>> eventSchedules = new ConcurrentHashMap<>();
    private Scheduler scheduler = null;
    private boolean started = false;

    public JobEventProduceService() throws SchedulerException {
        this.scheduler = schedulerFactory.getScheduler();
    }

    /**
     * 发起事件调度，返回事件调度的job列表
     *
     * @param group
     * @param event
     * @param parameters
     * @return
     */
    @Override
    public List<String> eventSchedule(String group, String event, String key, Map<String, String> parameters) {
        //获取关注此事件的所有schedule
        List<String> jobInfos = new ArrayList<>();
        Set<EventJobSchedule> schedules = eventSchedules.get(getMapKey(group, event));
        //获取schedule的job信息
        if (schedules != null && schedules.size() > 0) {
            for (JobSchedule schedule : schedules) {
                if (StringUtils.isNotEmpty(key)) {
                    schedule.setKey(key);
                }
                JobContext jobContext = new JobContext(schedule.getJobClass(), schedule);
                jobContext.setJobParameters(parameters);
                try {
                    boolean success = mqProducer.produce(jobContext, 0);
                    if (success) {
                        jobInfos.add(schedule.getJobClass());
                    }
                    logger.debug("发送任务调度消息:job:{}, schedule:{}, success:{}", jobContext.getJobClass(), jobContext.getSchedule(), success);
                } catch (Exception e) {
                    logger.error("将调度消息发送到消息队列失败. schedule:{}", schedule, e);
                }
            }
        }
        //发送调度消息
        return jobInfos;
    }

    /**
     * 发起异步任务
     *
     * @param group
     * @param jobClass
     * @param delayLevel
     * @param parameters
     * @return
     */
    @Override
    public String asyncSchedule(String group, String jobClass, String key, Integer delayLevel, Map<String, String> parameters) {
        JobSchedule schedule = asyncSchedules.get(getMapKey(group, jobClass));
        if (schedule == null) {
            logger.error("未能找到匹配的job调度信息对象！！！");
            return null;
        }
        if (StringUtils.isNotEmpty(key)) {
            schedule.setKey(key);
        }
        String job = null;
        //设置参数
        JobContext jobContext = new JobContext(schedule.getJobClass(), schedule);
        jobContext.setJobParameters(parameters);
        try {
            boolean success = mqProducer.produce(jobContext, delayLevel);
            if (success) {
                job = schedule.getJobClass();
            }
            logger.debug("发送任务调度消息:job:{}, schedule:{}, success:{}", jobContext.getJobClass(), jobContext.getSchedule(), success);
        } catch (Exception e) {
            logger.error("将调度消息发送到消息队列失败. schedule:{}", schedule, e);
        }
        return job;
    }

    private String getMapKey(String group, String name) {
        return String.format("%s#%s", group, name);
    }

    public synchronized void init() {
        //List<JobSchedule> schedules = registryCenter.getJobSchedules();
        //initSchedules(schedules);
        try {
            mqProducer.start();
        } catch (Exception e) {
            logger.error("produce服务连接rocketmq失败！", e);
        }
    }

    private synchronized void initSchedules(List<JobSchedule> schedules) {
        asyncSchedules.clear();
        eventSchedules.clear();
        if (schedules == null || schedules.size() == 0) {
            logger.warn("当前没有任何调度配置！");
            return;
        }
        for (JobSchedule schedule : schedules) {
            if (schedule.getType() == ScheduleType.ASYNC) {
                AsyncJobSchedule asyncJobSchedule = (AsyncJobSchedule) schedule;
                asyncSchedules.put(getMapKey(schedule.getGroup(), schedule.getJobClass()), asyncJobSchedule);
            } else if (schedule instanceof EventJobSchedule) {
                addEventJobSchedule((EventJobSchedule) schedule);
            }
        }
    }

    private synchronized void addEventJobSchedule(EventJobSchedule schedule) {
        EventJobSchedule eventJobSchedule = schedule;
        String key = getMapKey(eventJobSchedule.getGroup(), eventJobSchedule.getEvent());
        Set<EventJobSchedule> list = eventSchedules.get(key);
        if (list == null) {
            list = new CopyOnWriteArraySet<>();
            Set<EventJobSchedule> setList = eventSchedules.putIfAbsent(key, list);
            if (setList != null) {
                list = setList;
            }
        }
        list.add(eventJobSchedule);
    }

    @Override
    public void notifyScheduleAdded(JobSchedule schedule) {
        logger.debug("get schedule added notify. schedule:{}", schedule);
        //维护本地的schedule信息
        synchronized (this) {
            if (schedule instanceof EventJobSchedule) {
                addEventJobSchedule((EventJobSchedule) schedule);
            } else if (schedule instanceof AsyncJobSchedule) {
                AsyncJobSchedule asyncJobSchedule = (AsyncJobSchedule) schedule;
                asyncSchedules.put(getMapKey(schedule.getGroup(), schedule.getJobClass()), asyncJobSchedule);
            }
        }
        if (!started) {
            return;
        }
        if (!schedule.isEnabled() || !(schedule instanceof TriggerJobSchedule)) {
            //nothing to do
            return;
        }
        try {
            Date date = addSchedule((TriggerJobSchedule) schedule);
            logger.debug("the schedule is to run at {}", date);
        } catch (Exception e) {
            logger.error("add schedule to quartz failed.", e);
        }
    }

    /**
     * 添加job任务调度
     *
     * @param schedule
     * @return
     * @throws SchedulerException
     */
    public Date addSchedule(TriggerJobSchedule schedule) throws SchedulerException {
        if (scheduler != null && scheduler.isStarted()) {
            return submitJobSchedule(schedule);
        }
        return null;
    }

    private Date submitJobSchedule(TriggerJobSchedule triggerJobSchedule) {
        JobDataMap jobData = new JobDataMap();
        jobData.put("schedule", triggerJobSchedule);
        jobData.put("type", triggerJobSchedule.getType());
        try {
            JobDetail detail = JobBuilder.newJob(TriggerJobEventProduceJob.class).setJobData(jobData).withIdentity(triggerJobSchedule.getName(), triggerJobSchedule.getGroup()).build();
            ScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronScheduleNonvalidatedExpression(triggerJobSchedule.getCronExpression());
            Trigger trigger = TriggerBuilder.newTrigger().forJob(detail).withSchedule(scheduleBuilder).startNow().build();
            return scheduler.scheduleJob(detail, trigger);
        } catch (Exception e) {
            logger.error("添加对任务的调度失败. schedule name:{}, group:{}, cron:{}", triggerJobSchedule.getName(), triggerJobSchedule.getGroup(), triggerJobSchedule.getCronExpression(), e);
        }
        return null;
    }

    @Override
    public void notifyScheduleChanged(JobSchedule schedule) {
        logger.debug("get schedule change notify. schedule:{}", schedule);
        //TODO 维护本地的schedule信息，考虑到change复杂度较高，暂不实现，操作层面不提供修改接口，以删除然后新增代替
        if (!started) {
            return;
        }
        if (!(schedule instanceof TriggerJobSchedule)) {
            //nothing to do
            return;
        }
        try {
            changeSchedule((TriggerJobSchedule) schedule);
        } catch (SchedulerException e) {
            logger.error("update schedule error. schedule:{}", schedule, e);
        }
    }

    /**
     * 修改指定的调度任务，实现方案是先删除，再根据情况新增
     *
     * @param schedule
     * @return
     * @throws SchedulerException
     */
    public boolean changeSchedule(TriggerJobSchedule schedule) throws SchedulerException {
        if (scheduler != null && scheduler.isStarted()) {
            //判断修改类型，启用/停用/修改调度脚本 或者同时修改,简单的方法就是先把原来的删除，然后根据情况新建
            JobKey key = JobKey.jobKey(schedule.getName(), schedule.getGroup());
            boolean exists = scheduler.checkExists(key);

            if (exists) {
                //删除调度信息
                deleteSchedule(schedule.getGroup(), schedule.getName());
            }
            if (schedule.isEnabled()) {
                //修改调度脚本
                submitJobSchedule(schedule);
            }
            return true;
        }
        logger.warn("quartz scheduler is not running.");
        return false;
    }

    /**
     * 删除指定的调度任务信息
     *
     * @param group
     * @param name
     * @return
     * @throws SchedulerException
     */
    public boolean deleteSchedule(String group, String name) throws SchedulerException {
        if (scheduler != null && scheduler.isStarted()) {
            JobKey key = JobKey.jobKey(name, group);
            if (scheduler.checkExists(key)) {
                return scheduler.deleteJob(key);
            }
        }
        return false;
    }

    @Override
    public void notifyScheduleDeleted(JobSchedule schedule) {
        logger.debug("get schedule deleted notify. schedule group:{}, name:{}", schedule.getGroup(), schedule.getName());
        //维护本地的schedule信息
        synchronized (this) {
            if (schedule instanceof EventJobSchedule) {
                EventJobSchedule eventJobSchedule = (EventJobSchedule) schedule;
                String key = getMapKey(eventJobSchedule.getGroup(), eventJobSchedule.getEvent());
                Set<EventJobSchedule> list = eventSchedules.get(key);
                if (list != null) {
                    //遍历删除 group 和 name一致的对象
                    list.remove(schedule);
                }
            } else if (schedule instanceof AsyncJobSchedule) {
                String key = getMapKey(schedule.getGroup(), schedule.getJobClass());
                asyncSchedules.remove(key);
            }
        }
        if (!started) {
            return;
        }
        try {
            boolean result = deleteSchedule(schedule.getGroup(), schedule.getName());
            logger.debug("delete schedule from quartz, group:{}, name:{}, result:{}", schedule.getGroup(), schedule.getName(), result);
        } catch (SchedulerException e) {
            logger.error("delete schedule error.group:{}, name:{}", schedule.getGroup(), schedule.getName(), e);
        }
    }

    /**
     * 启动任务调度服务
     *
     * @throws Exception
     */
    public synchronized boolean start() {
        try {
            if (scheduler.isShutdown()) {
                scheduler = schedulerFactory.getScheduler();
            } else if (scheduler.isStarted()) {
                //重复启动
                return true;
            }
            //获取所有的调度信息
            List<JobSchedule> schedules = registryCenter.getJobSchedules();
            //目前只有定时调度的任务需要leader进行参与
            for (JobSchedule schedule : schedules) {
                if (schedule.isEnabled() && schedule.getType() == ScheduleType.TRIGGER) {
                    logger.debug("schedule:{}, 已经交由leader进行调度处理。", schedule.getName());
                    TriggerJobSchedule triggerJobSchedule = (TriggerJobSchedule) schedule;
                    submitJobSchedule(triggerJobSchedule);
                    //TODO 获取job的历史运行信息，确认是否需要额外的运行操作(比如失败重试等)
                }
            }
            initSchedules(schedules);
            scheduler.start();
            logger.info("启动任务事件发送服务");
            started = true;
            return true;
        } catch (Exception e) {
            logger.error("启动定时调度失败。", e);
        }
        return false;
    }

    /**
     * 停止任务调度服务的执行，此方法会等待所有未完成的任务完成后再返回
     *
     * @return
     */
    public synchronized boolean stop() {
        if (!started) {
            return true;
        }
        started = false;
        boolean result = true;
        try {
            scheduler.shutdown(true);
        } catch (Exception e) {
            logger.error("shutdown quartz schedule error.", e);
            result = false;
        }
        logger.info("停止任务调度服务的执行");
        return result;
    }
}
